package com.example.rocketmq.service;

import com.example.rocketmq.config.RocketMQProperties;
import com.example.rocketmq.model.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 限流消费者服务
 * 实现每分钟只消费10条消息的限流逻辑
 * 
 * @author demo
 */
@Slf4j
@Service
public class RateLimitConsumerService {

    @Autowired
    private RocketMQProperties properties;

    private DefaultMQPushConsumer consumer;
    
    // 限流计数器
    private final AtomicInteger consumedThisMinute = new AtomicInteger(0);
    private final AtomicLong totalConsumed = new AtomicLong(0);
    private final AtomicLong lastResetTime = new AtomicLong(System.currentTimeMillis());
    
    // 限流开关
    private volatile boolean rateLimitEnabled = true;

    @PostConstruct
    public void init() throws Exception {
        consumer = new DefaultMQPushConsumer(properties.getConsumer().getGroup());
        consumer.setNamesrvAddr(properties.getNamesrvAddr());
        consumer.subscribe(properties.getTopic().getName(), "*");
        
        // 消费模式使用默认值
        
        // 设置消息监听器
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                if (!rateLimitEnabled) {
                    log.warn("限流功能已禁用，快速消费所有消息");
                    // 快速消费所有消息
                    for (MessageExt msg : msgs) {
                        try {
                            processMessage(msg);
                            totalConsumed.incrementAndGet();
                        } catch (Exception e) {
                            log.error("消费消息失败，消息ID: {}", msg.getKeys(), e);
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                
                // 限流消费：控制消费速率，但不丢弃消息
                for (MessageExt msg : msgs) {
                    try {
                        // 根据配置选择限流策略
                        if (properties.getConsumer().getRateLimit().isWaitForNextMinute()) {
                            // 策略1：等待到下一分钟
                            while (isRateLimitExceeded()) {
                                log.info("已达到限流阈值，当前分钟已消费: {}/{}, 等待到下一分钟...", 
                                        consumedThisMinute.get(), 
                                        properties.getConsumer().getRateLimit().getMessagesPerMinute());
                                Thread.sleep(1000); // 等待1秒后再次检查
                            }
                        } else {
                            // 策略2：使用消费间隔控制速率
                            if (isRateLimitExceeded()) {
                                log.info("已达到限流阈值，当前分钟已消费: {}/{}, 使用消费间隔控制...", 
                                        consumedThisMinute.get(), 
                                        properties.getConsumer().getRateLimit().getMessagesPerMinute());
                                // 等待到下一分钟
                                while (isRateLimitExceeded()) {
                                    Thread.sleep(1000);
                                }
                            }
                        }
                        
                        // 消费消息
                        processMessage(msg);
                        consumedThisMinute.incrementAndGet();
                        totalConsumed.incrementAndGet();
                        
                        log.info("消息消费成功，消息ID: {}, 当前分钟已消费: {}/{}, 总计消费: {}", 
                                msg.getKeys(), consumedThisMinute.get(), 
                                properties.getConsumer().getRateLimit().getMessagesPerMinute(),
                                totalConsumed.get());
                        
                        // 消费间隔延迟（用于进一步控制消费速率）
                        if (properties.getConsumer().getRateLimit().getConsumeInterval() > 0) {
                            Thread.sleep(properties.getConsumer().getRateLimit().getConsumeInterval());
                        }
                        
                    } catch (Exception e) {
                        log.error("消费消息失败，消息ID: {}", msg.getKeys(), e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        
        consumer.start();
        log.info("限流消费者启动成功，NameServer: {}, 消费者组: {}, 限流配置: 每分钟{}条消息", 
                properties.getNamesrvAddr(), properties.getConsumer().getGroup(),
                properties.getConsumer().getRateLimit().getMessagesPerMinute());
    }

    @PreDestroy
    public void destroy() {
        if (consumer != null) {
            consumer.shutdown();
            log.info("限流消费者已关闭");
        }
    }

    /**
     * 处理消息
     */
    private void processMessage(MessageExt msg) {
        String messageBody = new String(msg.getBody());
        log.info("正在消费消息: Topic={}, Tags={}, Keys={}, Body={}", 
                msg.getTopic(), msg.getTags(), msg.getKeys(), messageBody);
        
        // 模拟业务处理
        try {
            // 模拟处理时间
            Thread.sleep(100);
            log.info("消息处理完成: {}", msg.getKeys());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("消息处理被中断: {}", msg.getKeys(), e);
        }
    }

    /**
     * 检查是否超过限流阈值
     */
    private boolean isRateLimitExceeded() {
        long currentTime = System.currentTimeMillis();
        long lastReset = lastResetTime.get();
        
        // 检查是否需要重置计数器（每分钟重置一次）
        if (currentTime - lastReset >= 60000) { // 60秒
            if (lastResetTime.compareAndSet(lastReset, currentTime)) {
                consumedThisMinute.set(0);
                log.info("重置限流计数器，新的一分钟开始");
            }
        }
        
        int limit = properties.getConsumer().getRateLimit().getMessagesPerMinute();
        return consumedThisMinute.get() >= limit;
    }

    /**
     * 每分钟重置限流计数器（备用方案）
     */
    @Scheduled(fixedRate = 60000)
    public void resetRateLimitCounter() {
        consumedThisMinute.set(0);
        log.info("定时重置限流计数器");
    }

    /**
     * 获取当前消费统计信息
     */
    public String getConsumptionStats() {
        long currentTime = System.currentTimeMillis();
        long lastReset = lastResetTime.get();
        long remainingTime = Math.max(0, 60000 - (currentTime - lastReset));
        
        return String.format("限流统计 - 当前分钟已消费: %d/%d, 总计消费: %d, 下次重置剩余时间: %d秒", 
                consumedThisMinute.get(), 
                properties.getConsumer().getRateLimit().getMessagesPerMinute(),
                totalConsumed.get(),
                remainingTime / 1000);
    }

    /**
     * 启用/禁用限流
     */
    public void setRateLimitEnabled(boolean enabled) {
        this.rateLimitEnabled = enabled;
        log.info("限流功能已{}", enabled ? "启用" : "禁用");
    }

    /**
     * 获取限流状态
     */
    public boolean isRateLimitEnabled() {
        return rateLimitEnabled;
    }

    /**
     * 获取当前分钟已消费消息数
     */
    public int getConsumedThisMinute() {
        return consumedThisMinute.get();
    }

    /**
     * 获取总计消费消息数
     */
    public long getTotalConsumed() {
        return totalConsumed.get();
    }
}
